package com.apps.sdses.flink141.source;

import com.apps.sdses.flink141.entity.VehicleDetection;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;

import java.text.DecimalFormat;
import java.text.SimpleDateFormat;
import java.util.Arrays;
import java.util.List;
import java.util.Random;
import java.util.concurrent.atomic.AtomicLong;

/**
 * @author jiwei
 * @description
 * @date 2023/5/31 9:37
 */
public class VehicleDetectionSource extends RichSourceFunction<VehicleDetection> {
    private static final Random random = new Random();
    private static final AtomicLong idGenerator = new AtomicLong(0);
    private static final DecimalFormat coordinateFormat = new DecimalFormat("0.000000");
    private static boolean runFlag = true;
    List<String> directions = Arrays.asList("东", "西", "南", "北");
    List<String> roads = Arrays.asList("经十路", "舜华南路", "旅游路");
    List<String> vehicleIds = Arrays.asList("车辆编号10000", "车辆编号10001", "车辆编号10002","车辆编号10003", "车辆编号10004", "车辆编号10005","车辆编号10006", "车辆编号10007", "车辆编号10008","车辆编号10009", "车辆编号100010");

    private static final SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss.SSS");


    @Override
    public void run(SourceContext<VehicleDetection> ctx) throws Exception {
        while (runFlag) {
            VehicleDetection vehicleDetection = new VehicleDetection();
            vehicleDetection.setId(idGenerator.getAndIncrement());
            double x = Double.parseDouble(coordinateFormat.format(random.nextDouble()));
            double y = Double.parseDouble(coordinateFormat.format(random.nextDouble()));
            vehicleDetection.setCoordinate("POINT(" + x + " " + y + ")");
            vehicleDetection.setDirection(directions.get(random.nextInt(directions.size())));
            vehicleDetection.setRoad(roads.get(random.nextInt(roads.size())));
            vehicleDetection.setVehicleId(vehicleIds.get(random.nextInt(vehicleIds.size())));
            vehicleDetection.setEventTime(sdf.format(System.currentTimeMillis()));
            ctx.collect(vehicleDetection);
            Thread.sleep(200);
        }
    }

    @Override
    public void cancel() {
        runFlag = false;
    }
}
